package com.rxjava;

import java.util.ArrayList;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

//zip：用来合并两个Observable发射的数据源，根据Func2函数生成一个新的值并发射出去，当其中一个Observable发送的数据结束或者出现异常后，另一个Observable也将停止发射数据
//zip是在其中一个Observable发射数据项后，组合所有Observable最早一个未被组合的数据项，也就是说，组合后的Observable发射的第n个数据项，必然是每个源由Observable各自发射的第n个数据项构成的。
public class ZipDemo {

	/**
	 * 模拟手机本地联系人查询
	 */
	private static Observable<List<String>> queryContactsFromLocation() {
	    return Observable.create(new ObservableOnSubscribe<List<String>>() {
	        @Override
	        public void subscribe(ObservableEmitter<List<String>> emitter) {
	        	System.out.println(Thread.currentThread().getName());
	        	try {
	                Thread.sleep(3000);
	            } catch (InterruptedException e) {
	                e.printStackTrace();
	            }

	            ArrayList<String> contacters = new ArrayList<>();
	            contacters.add(new String("location:张三"));
	            contacters.add(new String("location:李四"));
	            contacters.add(new String("location:王五"));
	            emitter.onNext(contacters);
	            emitter.onComplete();
	        }
	    });
	}

	/**
	 * 模拟网络联系人列表
	 */
	private static Observable<List<String>> queryContactsForNet() {
	    return Observable.create(new ObservableOnSubscribe<List<String>>() {
	        @Override
	        public void subscribe(ObservableEmitter<List<String>> emitter) {
	        	System.out.println(Thread.currentThread().getName());
	            try {
	                Thread.sleep(3000);
	            } catch (InterruptedException e) {
	                e.printStackTrace();
	            }

	            ArrayList<String> contacters = new ArrayList<>();
	            contacters.add(new String("net:Zeus"));
	            contacters.add(new String("net:Athena"));
	            contacters.add(new String("net:Prometheus"));
	            emitter.onNext(contacters);
	            emitter.onComplete();
	        }
	    });
	}

	public static void main(String[] args) {
		//当某界面内容来源不同，但需同时显示出来时
        //eg1:    一部分数据来自本地，一部分来自网络
		System.out.println(Thread.currentThread().getName());
		Observable.zip(
			queryContactsFromLocation(),
			queryContactsForNet(),
			new BiFunction<List<String>, List<String>, List<String>>() {
	            @Override
	            public List<String> apply(List<String> contacters, List<String> contacters2) {
	            	contacters.addAll(contacters2);
	                return contacters;
	            }
			}
		).subscribe(x -> System.out.println(x));

		Observable.zip(
			Observable.just(10, 20, 30),
			Observable.just(4, 8, 12, 16),
			new BiFunction<Integer, Integer, Integer>(){
				@Override
	            public Integer apply(Integer integer, Integer integer2) {
	                return integer + integer2;
	            }
			}
		).subscribe(x -> System.out.println(x));
	}

}
